
import { Channel, Connection, ConsumeMessage, connect } from 'amqplib';
export async function useMq() {
    const mqUrl = process.env.MQ_URL || ``
    const con = await connect(mqUrl)
    return con;
}

export async function useMqChannel() {
    const con = await useMq()
    const channel = await con.createChannel()
    return channel;
}

export async function publishToQueue(queueName: string, msg: string) {
    let con: Connection | undefined = undefined;
    let channel: Channel | undefined = undefined;
    try {
        con = await useMq()
        channel = await con.createChannel()
        await channel.assertQueue(queueName, {
            durable: true
        })
        channel.sendToQueue(queueName, Buffer.from(msg));
    } catch (e) {
        throw e;
    } finally {
        if (channel) await channel.close().catch(e => console.log(e))
        if (con) await con.close().catch(e => console.log(e))
    }
}

export async function publishToTopic(routingKey: string, message: string) {
    let con: Connection | undefined = undefined;
    let channel: Channel | undefined = undefined;
    try {
        // 建立与RabbitMQ的连接
        con = await useMq() // 请替换为你的RabbitMQ服务器地址
        // 创建一个通道
        channel = await con.createChannel();
        await channel.assertExchange(`amq.topic`, 'topic', { durable: true });
        // 发布消息到主题交换机，使用指定的路由键
        const result = channel.publish(`amq.topic`, routingKey, Buffer.from(message));

        console.log(`Sent exchange ${`amq.topic`} message "${message}" with routing key "${routingKey}" ${result}`);
    } catch (err) {
        console.error('Error occurred while publishing the message:', err);
    } finally {
        if (channel) await channel.close().catch(e => console.log(e))
        if (con) await con.close().catch(e => console.log(e))
    }
}

export interface IConsumeHandler {
    (msg: ConsumeMessage, channel: Channel): Promise<void>
}

export async function consumeTopicMessages(queueName: string, bindingKeys: string[], handler: IConsumeHandler): Promise<() => Promise<void>> {
    let connection: Connection | undefined;
    let channel: Channel | undefined;
    try {
        // 创建与RabbitMQ的连接
        connection = await useMq(); // 请根据实际情况替换连接字符串
        channel = await connection.createChannel();

        // 声明一个队列，这里使用一个随机的、排他的队列，也可以声明一个持久化的队列
        const queueResult = await channel.assertQueue(queueName, { exclusive: true });

        // 声明一个主题类型的交换机
        const exchangeName = `amq.topic`; // 交换机名称
        const ex = await channel.assertExchange(exchangeName, 'topic', { durable: true });

        // 绑定队列到交换机，使用指定的bindingKeys
        await Promise.all(bindingKeys.map(async bindingKey => {
            await channel!.bindQueue(queueResult.queue, ex.exchange, bindingKey);
        }));

        console.log(` [*] Waiting for messages on queue: ${queueResult.queue}. To exit press CTRL+C`);

        // 开始消费队列中的消息
        channel.consume(queueResult.queue, async (msg: ConsumeMessage | null) => {
            if (msg !== null) {
                console.log(` [x] Received routingKey: "${msg.fields.routingKey}", message: "${msg.content.toString()}"`);
                // 手动确认消息，根据需要可以调整为自动确认
                await handler(msg, channel!)
            } else {
                console.warn('Consumer canceled by server, shutting down');
                await connection!.close().catch(e => console.log(e));
            }
        }, { noAck: false }); // 设置为false以手动确认消息，true则自动确认
    } catch (err) {
        console.error('Error consuming messages:', err);
    }
    return async () => {
        if (channel) await channel.close().catch(e => console.log(e))
        if (connection) await connection.close().catch(e => console.log(e))
    }
}

export async function consumeMessages(queueName: string, handler: IConsumeHandler): Promise<void> {
    try {
        // 建立与RabbitMQ的连接
        const connection: Connection = await useMq(); // 请替换为你的RabbitMQ服务器地址
        // 创建一个通道
        const channel: Channel = await connection.createChannel();

        // 声明一个直连类型的队列，这里假设队列名为'hello'
        await channel.assertQueue(queueName, { durable: true });

        console.log(` [*] Waiting for messages in queue: ${queueName}. To exit press CTRL+C`);

        // 消费消息
        channel.consume(queueName, async (msg: ConsumeMessage | null) => {
            if (msg !== null) {
                console.log(` [x] Received: "${msg.content.toString()}"`);
                // 根据实际情况处理消息，然后确认消息已处理
                await handler(msg, channel)
            } else {
                // 消息为空时，通常是消费者被取消的信号
                console.warn('Consumer canceled by server, shutting down');
                await connection.close().catch(e => console.log(e));
            }
        }, { noAck: false }); // 设置noAck为false表示手动确认消息
    } catch (err) {
        console.error('Error occurred while consuming messages:', err);
    }
}
export async function bindTopicQueue(bindingKeys: string[], queueName: string) {
    let con: Connection | undefined = undefined;
    let channel: Channel | undefined = undefined;
    try {
        // 建立与RabbitMQ的连接
        con = await useMq() // 请替换为你的RabbitMQ服务器地址
        // 创建一个通道
        channel = await con.createChannel();
        await channel.assertExchange(`amq.topic`, 'topic', { durable: true });
        await channel.assertQueue(queueName, { durable: true });
        await Promise.all(bindingKeys.map(async bindingKey => {
            await channel!.bindQueue(queueName, `amq.topic`, bindingKey);
        }));
    } catch (e) {
        console.error('Error occurred while bind the queue:', e);
    } finally {
        if (channel) await channel.close().catch(e => console.log(e))
        if (con) await con.close().catch(e => console.log(e))
    }
}

